#pragma once
#include "requestor.h"
#include "rpc_caller.h"
#include "../common/dispather.h"
#include "rpc_topic.h"
#include "rpc_register.h"

namespace rpc
{
    namespace client
    {
        class RegisterClient
        {
        public:
            using ptr = std::shared_ptr<RegisterClient>;
            RegisterClient(const std::string &ip, int port)
                : _requestor(std::make_shared<Requestor>()), _provider(std::make_shared<client::Provider>(_requestor)), _dispather(std::make_shared<dispather>())
            {
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispather->registerHandler<BaseMessage>(MType::RSP_SERVICE, rsp_cb);

                auto message_cb = std::bind(&dispather::onMessage, _dispather.get(),
                                            std::placeholders::_1, std::placeholders::_2);
                _client = ClientFactory::create(ip, port);
                _client->setMessageCallback(message_cb);
                _client->connect();
            }
            bool registerMthod(const std::string &method, const Address &addr)
            {
                return _provider->registermethod(_client->connection(), method, addr);
            }

        private:
            Requestor::ptr _requestor;
            client::Provider::ptr _provider;
            dispather::ptr _dispather;
            BaseClient::ptr _client;
        };
        class DiscoveryClient
        {

        public:
            using ptr = std::shared_ptr<DiscoveryClient>;

            DiscoveryClient(const std::string &ip, int port, const client::Discover::offlineCallback &cb)
                : _requestor(std::make_shared<Requestor>()), _discover(std::make_shared<client::Discover>(_requestor, cb)),
                  _dispather(std::make_shared<dispather>())
            {
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispather->registerHandler<BaseMessage>(MType::RSP_SERVICE, rsp_cb);

                auto req_sb = std::bind(&client::Discover::onServiceRequest, _discover.get(), std::placeholders::_1, std::placeholders::_2);
                _dispather->registerHandler<ServiceRequest>(MType::REQ_SERVICE, req_sb);

                auto message_cb = std::bind(&dispather::onMessage, _dispather.get(), std::placeholders::_1, std::placeholders::_2);

                _client = ClientFactory::create(ip, port);
                _client->setMessageCallback(message_cb);
                _client->connected();
            }
            bool serviceDiscovery(const std::string &method, Address &addr)
            {
                return _discover->serviceDiscovery(_client->connection(), method, addr);
            }

        private:
            Requestor::ptr _requestor;
            client::Discover::ptr _discover;
            dispather::ptr _dispather;
            BaseClient::ptr _client;
        };
        class RpcClient
        {
        public:
            using ptr = std::shared_ptr<RpcClient>;
            RpcClient(bool enableDiscovery, const std::string ip, int port)
                : _enableDiscovery(enableDiscovery),
                  _requestor(std::make_shared<Requestor>()),
                  _dispather(std::make_shared<dispather>()),
                  _caller(std::make_shared<RpcCaller>(_requestor))
            {
                auto rsp_cb = std::bind(&client::Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispather->registerHandler<BaseMessage>(MType::REQ_RPC, rsp_cb);
                if (_enableDiscovery)
                {
                    auto offline_cb = std::bind(&RpcClient::delClient, this, std::placeholders::_1);
                    _discover_client = std::make_shared<DiscoveryClient>(ip, port, offline_cb);
                }
                else
                {
                    auto message_cb = std::bind(dispather::onMessage, _dispather.get(), std::placeholders::_1, std::placeholders::_2);
                    _rpc_client = ClientFactory::create(ip, port);
                    _rpc_client->setMessageCallback(message_cb);
                    _rpc_client->connect();
                }
            }
            bool call(const std::string &method, const Json::Value &params, Json::Value &result)
            {
                // 获取服务提供者：1. 服务发现；  2. 固定服务提供者
                BaseClient::ptr client = getclient(method);
                if (client.get() == nullptr)
                {
                    return false;
                }
                // 3. 通过客户端连接，发送rpc请求
                return _caller->call(client->connection(), method, params, result);
            }
            bool call(const std::string &method, const Json::Value &params, RpcCaller::JsonAsyncResponse &result)
            {
                BaseClient::ptr client = getclient(method);
                if (client.get() == nullptr)
                {
                    return false;
                }
                // 3. 通过客户端连接，发送rpc请求
                return _caller->call(client->connection(), method, params, result);
            }
            bool call(const std::string &method, const Json::Value &params, const RpcCaller::JsonResponseCallback &cb)
            {
                BaseClient::ptr client = getclient(method);
                if (client.get() == nullptr)
                {
                    return false;
                }
                // 3. 通过客户端连接，发送rpc请求
                return _caller->call(client->connection(), method, params, cb);
            }

        private:
            BaseClient::ptr newclient(const Address &host)
            {
                auto message_cb = std::bind(&dispather::onMessage, _dispather.get(), std::placeholders::_1, std::placeholders::_2);
                auto client = ClientFactory::create(host.first, host.second);
                client->setMessageCallback(message_cb);
                client->connect();
                putClient(host, client);
                return client;
            };
            BaseClient::ptr getclient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _rpc_clients.find(host);
                if (it == _rpc_clients.end())
                {
                    return BaseClient::ptr();
                }
                return it->second;
            }
            BaseClient::ptr getclient(const std::string &method)
            {
                BaseClient::ptr client;
                if (_enableDiscovery)
                {
                    Address host;
                    bool ret = _discover_client->serviceDiscovery(method, host);
                    if (ret == false)
                    {
                        ELOG("当前 %s 服务，没有找到服务提供者！", method.c_str());
                        return BaseClient::ptr();
                    }
                    client = getclient(host);
                    if (client.get() == nullptr)
                    {
                        client = newclient(host);
                    }
                }
                else
                {
                    client = _rpc_client;
                }
                return client;
            }
            void putClient(const Address &addr, BaseClient::ptr &client)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.insert(std::make_pair(addr, client));
            }
            void delClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _rpc_clients.erase(host);
            }

        private:
            struct AddressHash
            {
                size_t operator()(const Address &host) const
                {
                    std::string addr = host.first + std::to_string(host.second);
                    return std::hash<std::string>{}(addr);
                }
            };
            bool _enableDiscovery;
            DiscoveryClient::ptr _discover_client;
            Requestor::ptr _requestor;
            RpcCaller::ptr _caller;
            dispather::ptr _dispather;
            BaseClient::ptr _rpc_client;
            std::mutex _mutex;
            std::unordered_map<Address, BaseClient::ptr, AddressHash> _rpc_clients;
        };
        class TopicClient
        {
        public:
            using ptr = std::shared_ptr<TopicClient>;
            TopicClient(const std::string &ip, int potr)
                : _requestor(std::make_shared<Requestor>()),
                  _dispather(std::make_shared<dispather>()),
                  _topic_manager(std::make_shared<TopicManager>(_requestor))
            {
                auto rsp_cb = std::bind(&Requestor::onResponse, _requestor.get(), std::placeholders::_1, std::placeholders::_2);
                _dispather->registerHandler<BaseMessage>(MType::RSP_TOPIC, rsp_cb);

                auto msg_cb = std::bind(TopicManager::onpublish, _topic_manager.get(), std::placeholders::_1, std::placeholders::_2);
                _dispather->registerHandler<TopicRequest>(MType::REQ_TOPIC, msg_cb);
                auto message_cb = std::bind(&dispather::onMessage, _dispather.get(), std::placeholders::_1, std::placeholders::_2);
                _rpc_client = ClientFactory::create(ip, potr);
                _rpc_client->setMessageCallback(message_cb);
                _rpc_client->connect();
            }

        private:
            Requestor::ptr _requestor;
            TopicManager::ptr _topic_manager;
            dispather::ptr _dispather;
            BaseClient::ptr _rpc_client;
        };
    }
}